package com.cx.repository;

import com.cx.entity.RedisEntity;
import com.cx.service.RedisService;
import com.cx.service.RedisTemplateAdapter;
import com.cx.support.RedisBatch;
import com.cx.support.RedisBatchExecute;
import com.cx.utils.BeanHelper;
import com.cx.utils.ProtoStuffUtil;
import com.cx.wrapper.ObjWrapper;
import com.google.common.collect.Lists;
import com.querydsl.core.types.EntityPath;
import com.querydsl.core.types.OrderSpecifier;
import com.querydsl.core.types.Predicate;
import com.querydsl.core.types.dsl.PathBuilder;
import com.querydsl.jpa.JPQLQuery;
import com.querydsl.jpa.impl.AbstractJPAQuery;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.BooleanUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageImpl;
import org.springframework.data.domain.Pageable;
import org.springframework.data.domain.Sort;
import org.springframework.data.jpa.repository.support.*;
import org.springframework.data.querydsl.EntityPathResolver;
import org.springframework.data.querydsl.QSort;
import org.springframework.data.querydsl.QueryDslPredicateExecutor;
import org.springframework.data.querydsl.SimpleEntityPathResolver;
import org.springframework.data.redis.core.BoundHashOperations;
import org.springframework.data.redis.core.RedisOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.hash.BeanUtilsHashMapper;
import org.springframework.data.repository.support.PageableExecutionUtils;
import org.springframework.data.repository.support.PageableExecutionUtils.TotalSupplier;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;
import redis.clients.jedis.Pipeline;

import javax.persistence.EntityManager;
import javax.persistence.LockModeType;
import java.io.Serializable;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;

/**
 * QueryDsl specific extension of {@link SimpleJpaRepository} which adds implementation for
 * {@link QueryDslPredicateExecutor}.
 * 
 * @author Oliver Gierke
 * @author Thomas Darimont
 * @author Mark Paluch
 * @author Jocelyn Ntakpe
 * @author Christoph Strobl
 */
public class BaseQueryDslJpaRepository<T extends RedisEntity<ID>, ID extends Serializable> extends QueryDslJpaRepository<T, ID> {

	private static final EntityPathResolver DEFAULT_ENTITY_PATH_RESOLVER = SimpleEntityPathResolver.INSTANCE;

	private final EntityPath<T> path;
	private final PathBuilder<T> builder;
	private final Querydsl querydsl;

    private final RedisTemplate<String, ?> redisTemplate;

    private final BeanUtilsHashMapper<T> beanUtilsHashMapper;

    private final EntityManager em;

    private final Class<T> clazz;

	/**
	 * Creates a new {@link BaseQueryDslJpaRepository} from the given domain class and {@link EntityManager} and uses the
	 * given {@link EntityPathResolver} to translate the domain class into an {@link EntityPath}.
	 * 
	 * @param entityInformation must not be {@literal null}.
	 * @param entityManager must not be {@literal null}.
	 */
	public BaseQueryDslJpaRepository(JpaEntityInformation<T, ID> entityInformation, Class<T> domainClass, EntityManager entityManager) {
		super(entityInformation, entityManager);
		this.path = DEFAULT_ENTITY_PATH_RESOLVER.createPath(entityInformation.getJavaType());
		this.builder = new PathBuilder<T>(path.getType(), path.getMetadata());
		this.querydsl = new Querydsl(entityManager, builder);

        this.clazz = domainClass;
        this.em = entityManager;
        this.redisTemplate = RedisService.getRedisTemplate(false);
        beanUtilsHashMapper = new BeanUtilsHashMapper(domainClass);
	}

	/*
	 * (non-Javadoc)
	 * @see org.springframework.data.querydsl.QueryDslPredicateExecutor#findOne(com.mysema.query.types.Predicate)
	 */
	@Override
	public T findOne(Predicate predicate) {
        int conditionsHashcode = new String(ProtoStuffUtil.serialize(predicate)).hashCode();
        String idskey = BeanHelper.key(keyspace(), "findOne", new String[]{"predicate"}, new Object[]{conditionsHashcode});
        try {
            String entitykey = entityKey(idskey);
            if(StringUtils.isNotBlank(entitykey)) {
                T entity = getOnlyOne(entitykey);

                if (!ObjectUtils.isEmpty(entity)) {
                    return entity;
                }
            }

            final T t = createQuery(predicate).select(path).fetchOne();

            if(null == t) {
                return t;
            }

            BoundHashOperations<String, String, String> operations = redisTemplate.boundHashOps(key(t.getId()));
            BeanHelper.registerConvertUtils();
            Map<String, String> map = beanUtilsHashMapper.toHash(t);
            operations.putAll(map);

            RedisService.delete(idskey);
            RedisService.putObjCache(idskey, t.getId());

            return t;
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
	}

	/*
	 * (non-Javadoc)
	 * @see org.springframework.data.querydsl.QueryDslPredicateExecutor#findAll(com.mysema.query.types.Predicate)
	 */
	@Override
	public List<T> findAll(Predicate predicate) {
        return findAllByCache(predicate, null, null);
	}

	/*
	 * (non-Javadoc)
	 * @see org.springframework.data.querydsl.QueryDslPredicateExecutor#findAll(com.mysema.query.types.Predicate, com.mysema.query.types.OrderSpecifier<?>[])
	 */
	@Override
	public List<T> findAll(Predicate predicate, OrderSpecifier<?>... orders) {
        return findAllByCache(predicate, null, orders);
	}

	/* 
	 * (non-Javadoc)
	 * @see org.springframework.data.querydsl.QueryDslPredicateExecutor#findAll(com.mysema.query.types.Predicate, org.springframework.data.domain.Sort)
	 */
	@Override
	public List<T> findAll(Predicate predicate, Sort sort) {
        return findAllByCache(predicate, sort, null);
	}

	/* 
	 * (non-Javadoc)
	 * @see org.springframework.data.querydsl.QueryDslPredicateExecutor#findAll(com.mysema.query.types.OrderSpecifier[])
	 */
	@Override
	public List<T> findAll(OrderSpecifier<?>... orders) {
        return findAllByCache(null, null, orders);
	}

	/*
	 * (non-Javadoc)
	 * @see org.springframework.data.querydsl.QueryDslPredicateExecutor#findAll(com.mysema.query.types.Predicate, org.springframework.data.domain.Pageable)
	 */
	@Override
	public Page<T> findAll(Predicate predicate, Pageable pageable) {
        String[] paramnames = new String[2];
        Object[] paramvals = new Object[2];
        if(!ObjectUtils.isEmpty(predicate)){
            paramnames[0] = "predicate";
            paramvals[0] = new String(ProtoStuffUtil.serialize(predicate)).hashCode();
        }
        if(!ObjectUtils.isEmpty(pageable)){
            paramnames[1] = "pageable";
            paramvals[1] = new String(ProtoStuffUtil.serialize(pageable)).hashCode();
        }

        String idskey = BeanHelper.key(keyspace(), "findAll", paramnames, paramvals);
        try {
            List<String> entitykeys = entityKeys(idskey);
            if(!CollectionUtils.isEmpty(entitykeys)) {
                List<T> finalEntities = getSpecHashDataOfPipelined(entitykeys);

                if (!CollectionUtils.isEmpty(finalEntities) && !CollectionUtils.isEmpty(entitykeys)) {
                    return new PageImpl(finalEntities);
                }
            }

            Page<T> result = findAllJpa(predicate, pageable);

            if(CollectionUtils.isEmpty(result.getContent())){
                return result;
            }

            execMultiHashPipelined(result.getContent(), idskey);

            return result;
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
	}

    private Page<T> findAllJpa(Predicate predicate, Pageable pageable) {

        final JPQLQuery<?> countQuery = createCountQuery(predicate);
        JPQLQuery<T> query = querydsl.applyPagination(pageable, createQuery(predicate).select(path));

        return PageableExecutionUtils.getPage(query.fetch(), pageable, new TotalSupplier() {

            @Override
            public long get() {
                return countQuery.fetchCount();
            }
        });
    }

	/*
	 * (non-Javadoc)
	 * @see org.springframework.data.querydsl.QueryDslPredicateExecutor#count(com.mysema.query.types.Predicate)
	 */
	@Override
	public long count(Predicate predicate) {
        int conditionsHashcode = new String(ProtoStuffUtil.serialize(predicate)).hashCode();
        String idskey = BeanHelper.key(keyspace(), "count", new String[]{"count"}, new Object[]{conditionsHashcode});
        try {
            ObjWrapper<Long> objWrapper = (ObjWrapper<Long>)countKey(idskey);
            if(!ObjectUtils.isEmpty(objWrapper)) {
                return objWrapper.getData();
            }

            final long l = createQuery(predicate).fetchCount();

            RedisService.delete(idskey);
            RedisService.putObjCache(idskey, new ObjWrapper<Long>(l));

            return l;
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
	}

	/* 
	 * (non-Javadoc)
	 * @see org.springframework.data.querydsl.QueryDslPredicateExecutor#exists(com.mysema.query.types.Predicate)
	 */
	@Override
	public boolean exists(Predicate predicate) {
		return this.count(predicate) > 0;
	}

	/**
	 * Creates a new {@link JPQLQuery} for the given {@link Predicate}.
	 * 
	 * @param predicate
	 * @return the Querydsl {@link JPQLQuery}.
	 */
	protected JPQLQuery<?> createQuery(Predicate... predicate) {
        AbstractJPAQuery<?, ?> query = querydsl.createQuery(path).where(predicate);
        CrudMethodMetadata metadata = getRepositoryMethodMetadata();

        if (metadata == null) {
            return query;
        }

        LockModeType type = metadata.getLockModeType();
        query = type == null ? query : query.setLockMode(type);

        for (Entry<String, Object> hint : getQueryHints().entrySet()) {
            query.setHint(hint.getKey(), hint.getValue());
        }

        return query;
	}

	/**
	 * Creates a new {@link JPQLQuery} count query for the given {@link Predicate}.
	 *
	 * @param predicate, can be {@literal null}.
	 * @return the Querydsl count {@link JPQLQuery}.
	 */
	protected JPQLQuery<?> createCountQuery(Predicate predicate) {
        AbstractJPAQuery<?, ?> query = querydsl.createQuery(path).where(predicate);

        CrudMethodMetadata metadata = getRepositoryMethodMetadata();

        if (metadata == null) {
            return query;
        }

        for (Entry<String, Object> hint : metadata.getQueryHints().entrySet()) {
            query.setHint(hint.getKey(), hint.getValue());
        }

        return query;
	}

	/**
	 * Executes the given {@link JPQLQuery} after applying the given {@link OrderSpecifier}s.
	 * 
	 * @param query must not be {@literal null}.
	 * @param orders must not be {@literal null}.
	 * @return
	 */
	private List<T> executeSorted(JPQLQuery<T> query, OrderSpecifier<?>... orders) {
		return executeSorted(query, new QSort(orders));
	}

	/**
	 * Executes the given {@link JPQLQuery} after applying the given {@link Sort}.
	 * 
	 * @param query must not be {@literal null}.
	 * @param sort must not be {@literal null}.
	 * @return
	 */
	private List<T> executeSorted(JPQLQuery<T> query, Sort sort) {
		return querydsl.applySorting(sort, query).fetch();
	}

    private String keyspace(){
        return RedisTemplateAdapter.cacheLevel2Pre() + getDomainClass().getSimpleName();
    }

    private String key(ID id){
        return keyspace() + ":ids:" + id;
    }

    private ObjWrapper countKey(String key){
        Boolean hasKey = redisTemplate.hasKey(key);
        if(BooleanUtils.isFalse(hasKey)){
            return null;
        }

        return RedisService.getObjCache(key, ObjWrapper.class);
    }

    private String entityKey(String key){
        Boolean hasKey = redisTemplate.hasKey(key);
        if(BooleanUtils.isFalse(hasKey)){
            return null;
        }

        return RedisService.getObjCache(key, String.class);
    }

	private List<String> entityKeys(String key){
		Boolean hasKey = redisTemplate.hasKey(key);
		if(BooleanUtils.isFalse(hasKey)){
			return null;
		}

		return RedisService.getListCache(key, String.class);
	}

    private T getOnlyOne(String key){
        Boolean hasKey = redisTemplate.hasKey(key);
        if(BooleanUtils.isFalse(hasKey)){
            return null;
        }
        BoundHashOperations<String, String, String> operations = redisTemplate.boundHashOps(key);
        Map<String, String> entries = operations.entries();
        BeanHelper.registerConvertUtils();
        return beanUtilsHashMapper.fromHash(entries);
    }

    private List<T> getEntities(List<Object> result){
        List<T> list = Lists.newArrayListWithCapacity(result.size());
        result.stream().filter(p->null!=p).forEach(obj -> {
            Map<String, String> map = (Map<String, String>)obj;
            BeanHelper.registerConvertUtils();
            T entity = beanUtilsHashMapper.fromHash(map);
            if (Objects.nonNull(entity) && Objects.nonNull(entity.getId())) {
                list.add(entity);
            }
        });

        return list;
    }

    private List<T> findAllByCache(Predicate predicate, Sort sort, OrderSpecifier<?>... orders) {
        String[] paramnames = new String[3];
        Object[] paramvals = new Object[3];
        if(!ObjectUtils.isEmpty(predicate)){
            paramnames[0] = "predicate";
            paramvals[0] = new String(ProtoStuffUtil.serialize(predicate)).hashCode();
        }
        if(!ObjectUtils.isEmpty(sort)){
            paramnames[1] = "sort";
            paramvals[1] = new String(ProtoStuffUtil.serialize(sort)).hashCode();
        }
        if(!ObjectUtils.isEmpty(orders)){
            paramnames[2] = "orders";
            paramvals[2] = new String(ProtoStuffUtil.serialize(orders)).hashCode();
        }

        String idskey = BeanHelper.key(keyspace(), "findAll", paramnames, paramvals);
        try {
            List<String> entitykeys = entityKeys(idskey);
            if(!CollectionUtils.isEmpty(entitykeys)) {
                List<T> finalEntities = getSpecHashDataOfPipelined(entitykeys);

                if (!CollectionUtils.isEmpty(finalEntities) && !CollectionUtils.isEmpty(entitykeys)) {
                    return finalEntities;
                }
            }

            List<T> result = null;
            if(!ObjectUtils.isEmpty(predicate) && ObjectUtils.isEmpty(sort) && ArrayUtils.isEmpty(orders)){
                result = createQuery(predicate).select(path).fetch();
            }
            if(!ObjectUtils.isEmpty(predicate) && ObjectUtils.isEmpty(sort) && ArrayUtils.isNotEmpty(orders)){
                result = executeSorted(createQuery(predicate).select(path), orders);
            }
            if(!ObjectUtils.isEmpty(predicate) && !ObjectUtils.isEmpty(sort) && ArrayUtils.isEmpty(orders)){
                result = executeSorted(createQuery(predicate).select(path), sort);
            }
            if(ObjectUtils.isEmpty(predicate) && ObjectUtils.isEmpty(sort) && ArrayUtils.isNotEmpty(orders)){
                result = executeSorted(createQuery(new Predicate[0]).select(path), orders);
            }
            if(CollectionUtils.isEmpty(result)){
                return result;
            }

            execMultiHashPipelined(result, idskey);

            return result;
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }


    private List<T> getSpecHashDataOfPipelined(List<String> entitykeys){
        List<Object> result = RedisService.batchExecute(new RedisBatch() {
            @Override
            public void call(Pipeline pipeline) {
                entitykeys.forEach(key -> {
                    pipeline.hgetAll(key);
                });
            }
        });
        return getEntities(result);

//        List<Object> result = RedisService.executePipelined(new RedisBatchExecute() {
//            @Override
//            public void call(RedisOperations operations) {
//                entitykeys.forEach(key -> {
//                    BeanHelper.getOnlyOne(key, operations);
//                });
//            }
//        });
//        return getEntities(result);
    }

    private void execMultiHashPipelined(List<T> ts, String idskey){
//        RedisService.batchExecute(new RedisBatch() {
//            @Override
//            public void call(Pipeline pipeline) {
//                List<String> ids = Lists.newArrayListWithCapacity(10);
//                ts.forEach(t ->
//                        {
//                            ids.add(t.getId()+"");
//                            putAllToCache(pipeline, t);
//                        }
//                );
//
//                RedisService.delete(idskey);
//                RedisService.putListCache(idskey, ids);
//            }
//        });
        RedisService.execute(new RedisBatchExecute() {
            @Override
            public void call(RedisOperations operations) {
                List<String> ids = Lists.newArrayListWithCapacity(10);
                ts.forEach(t ->
                        {
                            ids.add(t.getId()+"");
                            putAllToCache(operations, t);
                        }
                );

                RedisService.delete(idskey);
                RedisService.putListCache(idskey, ids);
            }
        }, ts.hashCode()+"");
    }

//    private void putAllToCache(Pipeline pipeline, T t){
//        BeanHelper.registerConvertUtils();
//        Map<String, String> map = beanUtilsHashMapper.toHash(t);
//        pipeline.hmset(key(t.getId()), map);
//    }

    private void putAllToCache(RedisOperations operations, T t){
        BoundHashOperations<String, String, String> hashOperations = operations.boundHashOps(key(t.getId()));
        BeanHelper.registerConvertUtils();
        Map<String, String> map = beanUtilsHashMapper.toHash(t);
        hashOperations.putAll(map);
    }
}